package com.atguigu.gmall.realtime.dim.function;/**
 * @Author: Wxs
 * @Description: TODO
 * @Date: 2024/12/14 10:39
 */

import com.alibaba.fastjson.JSONObject;
import com.group2.edu.realtime.common.bean.TableProcessDim;
import com.group2.edu.realtime.common.constant.Constant;
import com.group2.edu.realtime.common.util.HBaseUtil;
import com.group2.edu.realtime.common.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * @Author: Wxs
 * @Description: TODO
 * @Date: 2024/12/14 10:39
 */
public class DimSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {
    private Connection hBaseConnection;
    private Jedis jedis;
    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseConnection = HBaseUtil.getConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeConnection(hBaseConnection);
        RedisUtil.closeJedis(jedis);
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> value, Context context) throws Exception {
        JSONObject dataJsonObj = value.f0;
        TableProcessDim tableProcessDim = value.f1;
        String type = dataJsonObj.getString("type");
        dataJsonObj.remove("type");
        String sinkTable = tableProcessDim.getSinkTable();
        String rowKey = dataJsonObj.getString(tableProcessDim.getSinkRowKey());
        if ("delete".equals(type)){
            //从hbase表中删除一条数据
            HBaseUtil.deleteData(hBaseConnection, Constant.HBASE_NAMESPACE,sinkTable,rowKey);

        }else{
            //将数据写入hbase,put操作
            String sinkFamily = tableProcessDim.getSinkFamily();
            HBaseUtil.insertData(hBaseConnection,Constant.HBASE_NAMESPACE,sinkTable,rowKey,sinkFamily,dataJsonObj);
        }
        //如果维度数据发生了变换,从redis中将缓存的数据清理掉
        if ("delete".equals(type) || "update".equals(type)){
            jedis.del(RedisUtil.getKey(sinkTable,rowKey));
        }


    }
}